package tech.powerjob.official.processors.impl.script;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Sets;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.SystemUtils;
import tech.powerjob.official.processors.CommonBasicProcessor;
import tech.powerjob.official.processors.util.CommonUtils;
import tech.powerjob.official.processors.util.JsonUtils;
import tech.powerjob.official.processors.util.ProcessUtils;
import tech.powerjob.worker.core.processor.ProcessResult;
import tech.powerjob.worker.core.processor.TaskContext;
import tech.powerjob.worker.log.OmsLogger;

import java.io.*;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * 脚本处理器
 *
 * @author tjq
 * @author Jiang Jining
 * @since 2020/4/16
 */
@Slf4j
public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
    protected static final String PYTHON_PROCESSOR = "python";
    protected static final String SHELL_PROCESSOR = "shell";

    // private static final ForkJoinPool POOL = new ForkJoinPool(/*4 * */Runtime.getRuntime().availableProcessors());
    private static final Set<String> DOWNLOAD_PROTOCOL = Sets.newHashSet("http", "https", "ftp");
    protected static final String SH_SHELL = "/bin/sh";
    private static final String WORKER_DIR = System.getProperty("user.home") + "/powerjob/worker/official_script_processor/";

    @Override
    protected ProcessResult process0(TaskContext context) throws Exception {
        OmsLogger omsLogger = context.getOmsLogger();
        String scriptParams = CommonUtils.parseParams(context);
        omsLogger.info("[SYSTEM] ScriptProcessor start to process, params: {}", scriptParams);
        if (scriptParams == null) {
            String message = "[SYSTEM] ScriptParams is null, please check jobParam configuration.";
            omsLogger.warn(message);
            return new ProcessResult(false, message);
        }
        String cdPath = null, scriptPath = null;
        List args = null;
        Map<String, Object> kwargs = null;
        // 如果是json参数
        if (JsonUtils.isJsonStr(scriptParams)) {
            try {
                AbstractScriptProcessor.JsonParams jsonParams = prepareExistsScriptFile(context.getInstanceId(), CommonUtils.parseParams(context));
                cdPath = jsonParams.getCd();
                scriptPath = jsonParams.getScriptPath();
                args = jsonParams.getArgs();
                kwargs = jsonParams.getKwargs();
            } catch (Exception e) {
                e.printStackTrace();
                return new ProcessResult(false, e.getMessage());
            }
            if (StringUtils.isBlank(scriptPath)) {
                return new ProcessResult(false, "jsonParams['scriptPath'] can't be empty!");
            }
        } else {
            scriptPath = prepareScriptFile(context.getInstanceId(), scriptParams);
        }

        omsLogger.info("[SYSTEM] Generate executable file successfully, path: {}", scriptPath);

        final List<String> runCommands = getRunCommand();
        if (SystemUtils.IS_OS_WINDOWS) {
            if (StringUtils.equals(runCommands.get(0), SH_SHELL)) {
                String message = String.format("[SYSTEM] Current OS is %s where shell scripts cannot run.", SystemUtils.OS_NAME);
                omsLogger.warn(message);
                return new ProcessResult(false, message);
            }
        }

        if(StringUtils.isNotBlank(cdPath)){
            // 防止同一个worker上并行执行同一个脚本任务。 // 检查脚本是否正在执行中，如果执行中直接到此结束
            final List<String> pids = ProcessUtils.getPids(Arrays.asList(scriptPath));
            if (null != pids && pids.size() > 0) {
                return new ProcessResult(false, "此脚本尚在执行中,如需重新执行,请先停止.");
            }
        }

        // 授权
        ProcessBuilder chmodPb = new ProcessBuilder("/bin/chmod", "755", scriptPath);
        // 等待返回，这里不可能导致死锁（shell产生大量数据可能导致死锁）
        chmodPb.start().waitFor();
        omsLogger.info("[SYSTEM] chmod 755 authorization complete.");

        // 获取执行命令的子进程输出信息: 正常的输出 和 错误的输出（子进程的输出，在主进程而言是输入）
        StringBuilder inputBuilder = new StringBuilder();
        StringBuilder errorBuilder = new StringBuilder();

        // 2. 执行目标脚本
        if(PYTHON_PROCESSOR == getScriptProcessorName()) {
            // 追加 执行脚本的 list、map/dict参数
            runCommands.set(2, String.format(runCommands.get(2), String.join(" ", Arrays.asList(scriptPath, JSON.toJSONString(args), JSON.toJSONString(kwargs)))));
        }else {
            // 追加 执行脚本的 list、map/dict参数
            runCommands.addAll(Arrays.asList(scriptPath, JSON.toJSONString(args), JSON.toJSONString(kwargs)));
        }

        ProcessBuilder pb = new ProcessBuilder(runCommands);
        // 切换工作目录
        if (StringUtils.isNotBlank(cdPath)) {
            pb.directory(new File(cdPath));
        }
        omsLogger.info("[SYSTEM] begin execution commands : " + String.join(" ", runCommands));

        // 合并错误流信息到标准输入流 pb.redirectErrorStream(true);
        // 執行命令, 返回一個子進程對象（命令在子進程中執行）
        Process process = pb.start();
        boolean success = false;
        String result;
        try {
            /*POOL.execute(new ForkJoinLog(process, inputBuilder, errorBuilder, omsLogger));*/
            // POOL.execute(() -> copyStream(process.getInputStream(), inputBuilder, omsLogger));
            // POOL.execute(() -> copyStream(process.getErrorStream(), errorBuilder, omsLogger));

            // waitFor为阻塞方法, 等待命令執行完成（成功會返回0）
            success = process.waitFor() == 0;
        } catch (InterruptedException ie) {
            omsLogger.info("[SYSTEM] ScriptProcessor has been interrupted");
        } finally {
            /* 销毁当前进程，阻断当前命令执行 process.destroy();*/
            ProcessUtils.killAllProcesses(Arrays.asList(scriptPath));
            // result = String.format("[INPUT]: %s;[ERROR]: %s", inputBuilder.toString(), errorBuilder.toString());
            result = String.format("task exec end");
        }
        return new ProcessResult(success, result);
    }


    /*public static class ForkJoinLog extends RecursiveAction {
        private boolean needFork;
        private OmsLogger omsLogger;
        private InputStream inputStream;
        private StringBuilder stringBuilder;

        private ForkJoinLog inputLogFork;
        private ForkJoinLog errLogFork;


        public ForkJoinLog(Process process, StringBuilder inputBuilder, StringBuilder errorBuilder, OmsLogger omsLogger) {
            this.needFork = true;
            this.inputLogFork = new ForkJoinLog(process.getInputStream(), inputBuilder, omsLogger);
            this.errLogFork = new ForkJoinLog(process.getErrorStream(), errorBuilder, omsLogger);
        }

        public ForkJoinLog(InputStream inputStream, StringBuilder stringBuilder, OmsLogger omsLogger) {
            this.needFork = false;
            this.inputStream = inputStream;
            this.stringBuilder = stringBuilder;
            this.omsLogger = omsLogger;
        }

        @Override
        protected void compute() {
            if (this.needFork) {
                inputLogFork.fork();
                errLogFork.fork();
                return;
            }

            String line;
            try (BufferedReader br = new BufferedReader(new InputStreamReader(this.inputStream, StandardCharsets.UTF_8))) {
                while ((line = br.readLine()) != null) {
                    this.stringBuilder.append(line);
                    // 同步到在线日志
                    this.omsLogger.info(line);
                }
            } catch (Exception e) {
                log.warn("[ScriptProcessor] copyStream failed.", e);
                this.omsLogger.warn("[SYSTEM] copyStream failed.", e);
                this.stringBuilder.append("Exception: ").append(e);
            }
        }
    }*/

    @Data
    public static class JsonParams {
        /**
         * 执行 已存在的脚本 需要切换到的路径
         */
        private String cd;
        /**
         * 已存在的脚本路径
         */
        private String scriptPath;
        /**
         * 已存在的脚本元祖参数
         */
        private List args;

        public List getArgs() {
            return null != args ? args : Collections.emptyList();
        }

        /**
         * 已存在的脚本字典参数
         */
        private Map<String, Object> kwargs;

        public Map<String, Object> getKwargs() {
            return null != kwargs ? kwargs : Collections.emptyMap();
        }
    }

    private AbstractScriptProcessor.JsonParams prepareExistsScriptFile(Long instanceId, String scriptParams) throws Exception {
        AbstractScriptProcessor.JsonParams res = JSON.parseObject(scriptParams, AbstractScriptProcessor.JsonParams.class);

        final File existsScriptFile = new File(res.getScriptPath());
        if (!existsScriptFile.exists()) {
            throw new RuntimeException("not found file by existsScriptPath: " + res.getScriptPath());
        }
        if (existsScriptFile.isDirectory()) {
            throw new RuntimeException("is not exec Script: " + res.getScriptPath());
        }
        final String scriptDirAbsolutePath = existsScriptFile.getParentFile().getAbsolutePath();
        if (StringUtils.isBlank(res.getCd())) {
            res.setCd(scriptDirAbsolutePath);
        }
        // 生成脚本文件的全路径
        // final String genScriptFilePath = Paths.get(scriptDirAbsolutePath, getScriptName(instanceId)).toString();
        // 重置执行脚本路径
        // res.setScriptPath(genScriptFilePath);
        // 复制脚本文件
        // try (FileInputStream is = new FileInputStream(existsScriptFile); FileChannel inChannel = is.getChannel();
        //      FileOutputStream os = new FileOutputStream(genScriptFilePath); FileChannel outChannel = os.getChannel()) {
        //     outChannel.transferFrom(inChannel, 0, inChannel.size());
        // } catch (Exception e) {
        //     e.printStackTrace();
        //     throw e;
        // }
        return res;
    }

    private String prepareScriptFile(Long instanceId, String processorInfo) throws IOException {
        String scriptPath = WORKER_DIR + getScriptName(instanceId);
        File script = new File(scriptPath);
        if (script.exists()) {
            return scriptPath;
        }
        File dir = new File(script.getParent());
        boolean success = dir.mkdirs();
        success = script.createNewFile();
        if (!success) {
            throw new RuntimeException("create script file failed");
        }

        // 如果是下载链接，则从网络获取
        for (String protocol : DOWNLOAD_PROTOCOL) {
            if (processorInfo.startsWith(protocol)) {
                FileUtils.copyURLToFile(new URL(processorInfo), script, 5000, 300000);
                return scriptPath;
            }
        }

        // 非下载链接，为 processInfo 生成可执行文件
        try (FileWriter fw = new FileWriter(script); BufferedWriter bw = new BufferedWriter(fw)) {
            bw.write(processorInfo);
            bw.flush();
        }
        return scriptPath;
    }

    private static void copyStream(InputStream is, StringBuilder sb, OmsLogger omsLogger) {
        String line;
        try (BufferedReader br = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
            while ((line = br.readLine()) != null) {
                // sb.append(line);
                // 同步到在线日志
                if (null != omsLogger) {
                    omsLogger.info(line);
                }
            }
        } catch (IOException e) {
            log.warn("[ScriptProcessor] copyStream failed.", e);
            if (null != omsLogger) {
                omsLogger.warn("[SYSTEM] copyStream failed.", e);
            }

            // sb.append("Exception: ").append(e.getMessage());
        }
    }

    protected abstract String getScriptProcessorName();
    /**
     * 生成脚本名称
     *
     * @param instanceId id of instance
     * @return 文件名称
     */
    protected abstract String getScriptName(Long instanceId);

    /**
     * 获取运行命令（eg，shell返回 /bin/sh）
     *
     * @return 执行脚本的命令
     */
    protected abstract List<String> getRunCommand();
}
